package cc.rengu.redp.hawk.service.impl;

import cc.rengu.redp.common.domain.RedpConstant;
import cc.rengu.redp.common.exception.RedisConnectException;
import cc.rengu.redp.common.redis.RedisService;
import cc.rengu.redp.common.utils.DateUtil;
import cc.rengu.redp.hawk.config.HawkConfig;
import cc.rengu.redp.hawk.domain.*;
import cc.rengu.redp.hawk.job.HawkPullJob;
import cc.rengu.redp.hawk.job.ScheduleQuietJob;
import cc.rengu.redp.hawk.service.HawkMonitorConfigService;
import cc.rengu.redp.hawk.service.HawkMonitorService;
import cc.rengu.redp.hawk.service.IndexArithmeticService;
import cc.rengu.redp.hawk.utils.HawkUtil;
import cc.rengu.redp.job.domain.Job;
import cc.rengu.redp.job.service.JobService;
import cc.rengu.redp.job.util.ScheduleUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.quartz.Scheduler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;


/**
 * @author xushitang
 */
@Slf4j
@Service
public class HawkMonitorServiceImpl implements HawkMonitorService {

    @Autowired
    private Scheduler scheduler;
    @Autowired
    private RedisService redisService;
    @Autowired
    private JobService jobService;
    @Autowired
    private HawkMonitorConfigService hawkMonitorConfigService;
    @Autowired
    private IndexArithmeticService indexArithmeticService;

    @Override
    public void push(String hawkId, String jsonStr) {
        JSONObject json = JSONObject.parseObject(jsonStr);
        this.push(hawkId, json);
    }

    @Override
    public void push(String hawkId, JSONObject json) {
        HawkMonitorConfigVo hawkMonitorConfigVo = hawkMonitorConfigService.getConfigById(hawkId);
        if (hawkMonitorConfigVo == null) {
            log.warn("hawkId:[{}], HawkMonitorConfig not found", hawkId);
            return;
        }

        if (!hawkMonitorConfigVo.getUseEnable()) {
            log.warn("hawkId:[{}], disabled", hawkId);
            return;
        }

        JSONObject indexes = this.dataToIndexes(hawkId, json, hawkMonitorConfigVo);
        if (indexes.size() != 0) {
            try {
                this.saveToRedis(hawkId, indexes, hawkMonitorConfigVo.getCachePeriod());
            } catch (RedisConnectException e) {
                log.error("push error", e);
            }
        }
    }

    @Override
    public Long createPullTask(HawkMonitorConfigVo hawkMonitorConfigVo) {
        Job job = new Job();
        job.setBeanName(HawkPullJob.class.getSimpleName());
        job.setMethodName(hawkMonitorConfigVo.getHawkId().toString()); //方法名填写监控ID
        job.setParams(hawkMonitorConfigVo.getCollectionUrl());
        job.setCronExpression("0/" + hawkMonitorConfigVo.getCollectionFreq() + " * * * * ? *");
        job.setCreateTime(new Date());
        job.setStatus(hawkMonitorConfigVo.getUseEnable() ? Job.ScheduleStatus.NORMAL.getValue() : Job.ScheduleStatus.PAUSE.getValue());
        if (jobService.save(job)) {
            ScheduleUtils.createScheduleJob(scheduler, job, ScheduleQuietJob.class);
            return job.getJobId();
        } else {
            return null;
        }
    }

    @Override
    public void updatePullTask(HawkMonitorConfigVo hawkMonitorConfigVo) {
        Job job = jobService.getById(hawkMonitorConfigVo.getCollectionJobId());
        job.setParams(hawkMonitorConfigVo.getCollectionUrl());
        job.setCronExpression("0/" + hawkMonitorConfigVo.getCollectionFreq() + " * * * * ? *");
        job.setStatus(hawkMonitorConfigVo.getUseEnable() ? Job.ScheduleStatus.NORMAL.getValue() : Job.ScheduleStatus.PAUSE.getValue());
        if (jobService.updateById(job)) {
            ScheduleUtils.updateScheduleJob(scheduler, job);
        }
    }

    @Override
    public void deletePullTask(Long collectionJobId) {
        if (jobService.removeById(collectionJobId)) {
            ScheduleUtils.deleteScheduleJob(scheduler, collectionJobId);
        }
    }

    @Override
    public void switchPullTask(Long collectionJobId, boolean enable) {
        Job job = jobService.getById(collectionJobId);
        job.setStatus(enable ? Job.ScheduleStatus.NORMAL.getValue() : Job.ScheduleStatus.PAUSE.getValue());
        if (jobService.updateById(job)) {
            ScheduleUtils.updateScheduleJob(scheduler, job);
        }
    }

    @Override
    public GraphData getData(GraphDataQuery graphDataQuery) {
        GraphData graphData = new GraphData();
        graphData.setHawkId(graphDataQuery.getHawkId());
        Long maxTimeStamp = null;

        HawkMonitorConfigVo hawkMonitorConfigVo = hawkMonitorConfigService.getConfigById(graphDataQuery.getHawkId());
        if (hawkMonitorConfigVo == null) {
            log.warn("can not get hawkMonitorConfigVo. hawkId:[{}]", graphDataQuery.getHawkId());
            return graphData;
        }

        if (!hawkMonitorConfigVo.getUseEnable() || !hawkMonitorConfigVo.getShowEnable()) {
            return graphData;
        }

        //线性图x轴（时间轴），径向图标签
        List<String> xAxisData = new ArrayList<>();
        //指标集
        Map<String, List<Float>> indexMap = new HashMap<>();

        //redis存储的数据，redisService返回的是有序的LinkedHashSet
        Set<String> redisData;
        try {
            redisData = redisService.zrangeByScore(
                    RedpConstant.HAWK_REALTIME_ZSET_PREFIX + graphDataQuery.getHawkId(),
                    "-INF",
                    "+INF");
            maxTimeStamp = dataOfAll(redisData, xAxisData, indexMap);
        } catch (Exception e) {
            log.error("get graph data error", e);
            return graphData;
        }

        filterIndexes(indexMap, hawkMonitorConfigVo.getIndexList());

        /**
         * 图形数据组装
         */
        if (HawkMonitorConfig.SHOW_CHART_LINE.equals(graphDataQuery.getShowChart())) {
            // 线性图
            List<LineChartDataSeries> lineChartDataSeriesList = indexMap.entrySet().stream().map(e -> {
                LineChartDataSeries dataSeries = new LineChartDataSeries();
                dataSeries.setName(getIndexDesc(e.getKey(), hawkMonitorConfigVo.getIndexList()));
                dataSeries.setData(e.getValue().toArray(new Float[e.getValue().size()]));
                return dataSeries;
            }).collect(Collectors.toList());

            graphData.setLineChartDataSeries(lineChartDataSeriesList);
            graphData.setLineChartXData(xAxisData.toArray(new String[xAxisData.size()]));
        } else if (HawkMonitorConfig.SHOW_CHART_RADIAL.equals(graphDataQuery.getShowChart())) {
            //径向图
            List<Float> dataSeries = new ArrayList<>();
            List<String> labels = new ArrayList<>();
            for (Map.Entry<String, List<Float>> e : indexMap.entrySet()) {
                labels.add(getIndexDesc(e.getKey(), hawkMonitorConfigVo.getIndexList()));
                if (e.getValue().size() != 0) {
                    //每次取最后（新）一项
                    dataSeries.add(e.getValue().get(e.getValue().size() - 1));
                } else {
                    dataSeries.add(0.0f);
                }
            }

            graphData.setRadialBarDataSeries(dataSeries.toArray(new Float[dataSeries.size()]));
            graphData.setRadialBarLabel(labels.toArray(new String[labels.size()]));
        } else if (HawkMonitorConfig.SHOW_CHART_LIST.equals(graphDataQuery.getShowChart())) {
            //列表图（用配置遍历，目的是为了保持与配置的指标顺序一致）
            List<ListChartData> listChartDatas = new ArrayList<>();
            hawkMonitorConfigVo.getIndexList().stream().forEach(e -> {
                List<Float> value = indexMap.get(e.getIndexName());
                if (value == null) return;
                ListChartData listChartData = new ListChartData();
                listChartData.setName(getIndexDesc(e.getIndexName(), hawkMonitorConfigVo.getIndexList()));
                if (value.size() > 0) {
                    listChartData.setValue(value.get(0));
                } else {
                    listChartData.setValue(0.0f);
                }
                listChartDatas.add(listChartData);
            });

            graphData.setListChartData(listChartDatas);
        }

        if (maxTimeStamp != null) {
            graphData.setMaxTimeStamp(maxTimeStamp.toString());
        }
        return graphData;
    }

    /**
     * 获取指标描述
     * @param indexName 指标名
     * @param indexList
     * @return
     */
    private String getIndexDesc(String indexName, List<HawkIndex> indexList) {
        Optional<HawkIndex> opt = indexList.stream().filter(i -> i.getIndexName().equals(indexName)).findFirst();
        if (opt.isPresent()) {
            if (StringUtils.isNotBlank(opt.get().getUnit())) {
                return opt.get().getIndexDescription() + "（" + opt.get().getUnit() + "）";
            } else {
                return opt.get().getIndexDescription();
            }
        } else {
            return indexName;
        }
    }

    /**
     * 过滤指标集
     * @param indexMap 从redis获取的指标集
     * @param hawkIndexList 指标配置列表
     */
    private void filterIndexes(Map<String, List<Float>> indexMap, List<HawkIndex> hawkIndexList) {
        hawkIndexList.stream().forEach(i -> {
            if (HawkIndex.HIDE_INDEX.equals(i.getShow())) {
                indexMap.remove(i.getIndexName());
            }
        });
    }

    /**
     * 把采集指标保存到redis
     * @param hawkId
     * @param indexes
     * @throws RedisConnectException
     */
    private void saveToRedis(String hawkId, JSONObject indexes, Integer keepAliveSeconds) throws RedisConnectException {
        long currentTimeMillis = System.currentTimeMillis();
        indexes.put(HawkIndex.TIME_TAG, currentTimeMillis);
        long expireTimeMillis = currentTimeMillis - keepAliveSeconds * 1000;
        redisService.zadd(RedpConstant.HAWK_REALTIME_ZSET_PREFIX + hawkId, (double) currentTimeMillis, indexes.toJSONString());
        //清理超过保留时长的记录
        redisService.zremrangeByScore(RedpConstant.HAWK_REALTIME_ZSET_PREFIX + hawkId, "-inf", Long.toString(expireTimeMillis));
    }

    /**
     * 采集指标
     * @param hawkId
     * @param dataJson
     * @return
     */
    private JSONObject dataToIndexes(String hawkId, JSONObject dataJson, HawkMonitorConfigVo hawkMonitorConfigVo) {
        JSONObject indexJson = new JSONObject();

        //用于计算型指标的历史数据
        Map<String, List<TimestampData>> indexesTimestampData = new HashMap<>();
        if (hawkMonitorConfigVo.getIndexList().stream()
                .filter(i -> HawkIndex.INDEX_TYPE_NORMAL.equals(i.getIndexType()))
                .count() > 0) {
            //含有计算指标，需获取所有历史数据
            try {
                Set<String> hisStringData = redisService.zrangeByScore(
                        RedpConstant.HAWK_REALTIME_ZSET_PREFIX + hawkId, "-INF", "+INF");
                if (hisStringData != null) {
                    List<JSONObject> hisData = hisStringData.stream().map(JSONObject::parseObject).collect(Collectors.toList());
                    indexesTimestampData.putAll(this.cacheHistoryDataConvertor(hisData));
                }
            } catch (RedisConnectException e) {
                log.error("get history data from redis error.");
                return indexJson;
            }
        }

        //遍历指标
        hawkMonitorConfigVo.getIndexList().forEach(i -> {
            if (HawkIndex.INDEX_TYPE_NORMAL.equals(i.getIndexType())) {
                /**
                 * 普通型指标
                 */
                Float indexValue = HawkUtil.jsonObjectValueToFloat(dataJson.get(i.getIndexName()));
                if (indexValue == null) {
                    indexValue = 0.0f;
                }
                indexJson.put(i.getIndexName(), indexValue);
            } else if (HawkIndex.INDEX_TYPE_ALGORITHM.equals(i.getIndexType())) {
                /**
                 * 计算型指标
                 */
                JSONObject algParamsJson = JSONObject.parseObject(i.getAlgParams());
                Float indexValue = indexArithmeticService.run(
                        i.getAlgorithm(),
                        algParamsJson,
                        dataJson,
                        indexesTimestampData);
                if (indexValue == null) {
                    indexValue = 0.0f;
                }
                indexJson.put(i.getIndexName(), indexValue);
            } else {
                log.error("incorrect index type : [{}]", i.getIndexType());
            }
        });

        return indexJson;
    }

    /**
     * 解析全量数据
     *
     * @param redisData redis返回的数据（输入）
     * @param xAxisData x轴数据 （输出）
     * @param indexMap  图表数据 （输出）
     * @return 最大时间戳
     *
     */
    private Long dataOfAll(Set<String> redisData, List<String> xAxisData, Map<String, List<Float>> indexMap) {
        int ignore = 0;
        Long maxTimeStamp = null;
        for (String data : redisData) {
            if (redisData.size() - ignore > HawkConfig.SHOW_X_AXIS_COUNT) {
                //只取最新的 HawkConfig.SHOW_X_AXIS_COUNT 条记录
                ignore++;
                continue;
            }

            JSONObject json = JSON.parseObject(data);
            for (Map.Entry index : json.entrySet()) {
                if (index.getKey().equals(HawkIndex.TIME_TAG)) {
                    //时间戳作为x轴
                    maxTimeStamp = Long.valueOf((long)index.getValue());
                    String time = DateUtil.getDateFormat(new Date((long) index.getValue()), DateUtil.TIME_PATTERN);
                    xAxisData.add(time);
                } else {
                    //指标
                    List<Float> indexData = indexMap.get(index.getKey());
                    if (indexData == null) {
                        indexData = new ArrayList<>();
                        indexMap.put((String) index.getKey(), indexData);
                    }
                    indexData.add(((BigDecimal) index.getValue()).floatValue());
                }
            }
        }

        return maxTimeStamp;
    }

    /**
     * 把缓存数据从json形式转化为key（指标名）-value（按时间排序的指标值）形式
     * @param hisData 示例：[{"hawkTime":"time1", "indexA":123, "indexB":456}, {"hawkTime":"time2", "indexA":789, "indexB":120}, ...]
     * @return 示例：[{key:"indexA", value:"[{timestamp:time1, indexData:123}, {timestamp:time2, indexData:789}]"},
     *               {key:"indexB", value:"[{timestamp:time1, indexData:456}, {timestamp:time2, indexData:120}]"},
     *               ...]
     */
    private Map<String, List<TimestampData>> cacheHistoryDataConvertor(List<JSONObject> hisData) {
        // 按时间排序
        hisData.stream().sorted(Comparator.comparing(i -> i.getLong(HawkIndex.TIME_TAG)));

        Map<String, List<TimestampData>> indexesTimestampData = new HashMap<>();
        //遍历历史数据中的每一条记录
        hisData.stream().forEach(d -> {
            Long timestamp = d.getLong(HawkIndex.TIME_TAG);
            if (timestamp == null) {
                log.warn("timestamp gotten from redis is null.");
                return;
            }
            //遍历每一条历史数据的各个指标
            d.entrySet().stream().forEach(e -> {
                if (e.getKey().equals(HawkIndex.TIME_TAG)) {
                    return;
                }
                List<TimestampData> indexTimestampData = indexesTimestampData.get(e.getKey());
                if (indexTimestampData == null) {
                    indexTimestampData = new ArrayList<>();
                    indexesTimestampData.put(e.getKey(), indexTimestampData);
                }
                TimestampData timestampData = new TimestampData();
                timestampData.setTimestamp(timestamp);
                timestampData.setIndexData(HawkUtil.jsonObjectValueToFloat(e.getValue()));
                indexTimestampData.add(timestampData);
            });
        });

        return indexesTimestampData;
    }
}
